package com.windy.kafka.trace.interceptor;

import com.windy.common.transmit.TransmitHeader;
import com.windy.kafka.trace.consumer.MessageConsumerProcessor;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.springframework.aop.framework.ProxyFactoryBean;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.kafka.listener.MessageListenerContainer;

import java.util.List;

/**
 * Created by liyufeng on 2022/6/3
 */
@Aspect
public class KafkaListenerAspect {

    private TransmitHeader transmitHeader;

    private List<MessageConsumerProcessor> messageConsumerProcessors;

    public KafkaListenerAspect(TransmitHeader transmitHeader,List<MessageConsumerProcessor> messageConsumerProcessors){
        this.transmitHeader = transmitHeader;
        this.messageConsumerProcessors = messageConsumerProcessors;
    }

    @Pointcut("execution (public * org.springframework.kafka.config.KafkaListenerContainerFactory.createListenerContainer(..))")
    private void createListenContainer(){};

    @Pointcut("execution (public * org.springframework.kafka.config.KafkaListenerContainerFactory.createContainer(..))")
    private void createContainer(){};


    @Around("createListenContainer()||createContainer()")
    public Object wrapListenerContainerWhenCreated(ProceedingJoinPoint joinPoint) throws Throwable {
        MessageListenerContainer listenerContainer = (MessageListenerContainer) joinPoint.proceed();
        if(listenerContainer instanceof AbstractMessageListenerContainer){
            AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) listenerContainer;
            Object messageListener = container.getContainerProperties().getMessageListener();
            if(messageListener instanceof MessageListener){
                container.setupMessageListener(creteProxy(messageListener));
            }
        }
        return listenerContainer;
    }

    private Object creteProxy(Object messageListener) {
        ProxyFactoryBean factoryBean = new ProxyFactoryBean();
        factoryBean.setProxyTargetClass(true);
        factoryBean.setTarget(messageListener);
        factoryBean.addAdvice(new MessageListenerMethodInterceptor(transmitHeader,messageConsumerProcessors));
        return factoryBean.getObject();
    }

}
